package com.atguigu.gmall.realtime.utils;

import com.atguigu.gmall.realtime.bean.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;

/*
操作ClickHouse的工具类
 */
public class ClickHouseUtil {
    //获取针对ClickHouse的JdbcSink

    public static <T> SinkFunction getJdbcSink(String sql){
        SinkFunction<T> sink = JdbcSink.<T>sink(
                sql,
                //t为数据对象，要装配到预编译器的参数中
                (jdbcPreparedStatement,t) ->{
                    Field[] fields = t.getClass().getDeclaredFields();
                    int skipOffset =0;
                    for (int i=0;i<fields.length;i++){
                        Field field = fields[i];
                        // 通过反射获取字段上的注解
                        TransientSink transientSink = field.getAnnotation(TransientSink.class);
                        if(transientSink != null){
                            //如果存在该注解
                            System.out.println("跳过字段： "+ field.getName());
                            skipOffset ++;
                            continue;
                        }
                        field.setAccessible(true);
                        try{
                            Object o = field.get(t);
                            //i代表流对象字段的下标
                            jdbcPreparedStatement.setObject(i+1 - skipOffset,o);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }

                },
                new JdbcExecutionOptions.Builder().withBatchSize(2).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(GmallConfig.CLICKHOUSE_URL)
                .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                .build()
        );
        return sink;
    }
}
